﻿using RabbitMQ.Client;
using RabbitMQ示例.单元测试;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Windows;
using System.Windows.Controls;
using System.Windows.Data;
using System.Windows.Documents;
using System.Windows.Input;
using System.Windows.Media;
using System.Windows.Media.Imaging;
using System.Windows.Navigation;
using System.Windows.Shapes;

namespace RabbitMQ示例
{
    /// <summary>
    /// MainWindow.xaml 的交互逻辑
    /// </summary>
    public partial class MainWindow : Window
    {
        public MainWindow()
        {
            InitializeComponent();


        }
        /// <summary>
        /// 
        /// </summary>
        public void 示例数据()
        {
            //重要说明,如果出现TCP联接异常,需要对所有联接相关的对像进行初始化,及设置

            IConnection conn;

            var factory = new RabbitMQ.Client.ConnectionFactory()
            {
                //出故障后自动恢复
                AutomaticRecoveryEnabled = true,
                //拓扑恢复
                TopologyRecoveryEnabled = true,
                //联接超时时间
                //RequestedConnectionTimeout=5,
                //超时重联时间间隔 2秒
                NetworkRecoveryInterval = new TimeSpan(0, 0, 2),
            };
            factory.HostName = "172.16.15.1";
            //factory.HostName = "192.168.2.110";
            //factory.HostName = "192.168.0.104";
            factory.UserName = "rabbitmq";
            factory.Password = "rabbitmq";
            factory.VirtualHost = "/";
            factory.ContinuationTimeout = factory.NetworkRecoveryInterval;

            #region 发布数据
            //开始发布数据
            System.Threading.Tasks.Task.Factory.StartNew(() =>
            {
                long ina = 0;

                //如果用户名或者密码出错  {"ACCESS_REFUSED - Login was refused using authentication mechanism PLAIN. For details see the broker logfile."}
                using (conn = factory.CreateConnection())
                {
                    //创建一个聊天记录
                    using (var channel = conn.CreateModel())
                    {
                        //设置超时时间
                        //channel.ContinuationTimeout = factory.ContinuationTimeout;
                        //设置属性
                        Topic类型(channel);
                        //Direct类型(channel);
                        //Fanout类型(channel);

                        //设置消息持久
                        var property = channel.CreateBasicProperties();
                        property.DeliveryMode = 2;

                        //是否联接中断过
                        bool isBroken = false;


                        for (int k = 0; k < 1000; k++)
                        {
                            for (int i = 0; i < 10; i++)
                            {
                                try
                                {
                                    //检查当前状态是否为打开状态
                                    if (channel.IsOpen)
                                    {
                                        //初始化联接等待一段时间防止发布时数据丢失
                                        if (isBroken)
                                        {
                                            System.Threading.Thread.Sleep(2 * 1000);
                                            //Topic类型(channel);
                                            isBroken = false;
                                        }
                                        System.Threading.Interlocked.Increment(ref ina);
                                        string message = "Hello World " + Guid.NewGuid() + " " + ina;
                                        var body = Encoding.UTF8.GetBytes(message);
                                        channel.BasicPublish("INCOM.DGSExchange", "DGS.1", property, body);
                                    }
                                    else
                                    {
                                        if (!isBroken)
                                        {
                                            isBroken = true;
                                            Console.WriteLine("发送联接断开..{0}", DateTime.Now);
                                        }
                                        System.Threading.Thread.Sleep(1000);
                                    }
                                }
                                //远程主机关闭异常
                                catch (RabbitMQ.Client.Exceptions.AlreadyClosedException)
                                {

                                }
                            }
                            System.Threading.Thread.Sleep(1000);
                        }


                        Console.WriteLine(" set {0}", "发送消息");
                    }
                }
            });
            #endregion


            #region 读取数据
            //开始发布数据
            System.Threading.Tasks.Task.Factory.StartNew(() =>
            {

                //如果用户名或者密码出错  {"ACCESS_REFUSED - Login was refused using authentication mechanism PLAIN. For details see the broker logfile."}
                using (var connection = factory.CreateConnection())
                {
                    //创建一个聊天记录
                    using (var channel = connection.CreateModel())
                    {
                        //交换机
                        string exchange = "INCOM.DGSExchange";
                        //队列
                        string queue = "INCOM.DGS1";
                        //声明exchange
                        channel.ExchangeDeclare(exchange, ExchangeType.Topic);
                        //声明queue //
                        channel.QueueDeclare(
                            //队列名
                            queue,
                            //是否持久化 durable
                            false,
                            //排它 exclusive
                            false,
                            //自动删除 autoDelete
                            false,
                            null);
                        //对exchange 与queue邦定
                        channel.QueueBind(queue, exchange, "DGS.*");

                        //是否联接中断过
                        bool isBroken = false;

                        #region 普通读取方式 
                        //普通使用方式BasicGet
                        //noAck = true，不需要回复，接收到消息后，queue上的消息就会清除
                        //noAck = false，需要回复，接收到消息后，queue上的消息不会被清除，直到调用channel.basicAck(deliveryTag, false); queue上的消息才会被清除 而且，在当前连接断开以前，其它客户端将不能收到此queue上的消息
                        BasicGetResult res = channel.BasicGet(queue, true);
                        if (res != null)
                        {
                            bool t = res.Redelivered;
                            t = true;
                            Console.WriteLine(System.Text.UTF8Encoding.UTF8.GetString(res.Body));
                            //返回消息
                            //channel.BasicAck(res.DeliveryTag, false);
                        }

                        #endregion

                        QueueingBasicConsumer consurment = new QueueingBasicConsumer(channel);
                        //设置消费者
                        var model = channel.BasicConsume(queue, true, consurment);
                        RabbitMQ.Client.Events.BasicDeliverEventArgs abc;

                        while (consurment.Model.IsOpen || factory.AutomaticRecoveryEnabled)
                        {
                            try
                            {
                                //判断是否正在运行
                                if (consurment.IsRunning)
                                {
                                    //初始化联接故障状态用于日志记录
                                    //if (isBroken)
                                    //{
                                    //     isBroken = false;
                                    //}
                                    abc = consurment.Queue.Dequeue();
                                    if (abc != null)
                                    {
                                        Console.WriteLine(System.Text.UTF8Encoding.UTF8.GetString(abc.Body));
                                    }

                                }
                                else
                                {
                                    //if (!isBroken)
                                    //{
                                    //    isBroken = true;
                                    //    Console.WriteLine("读取联接断开..{0}", DateTime.Now);
                                    //}
                                    System.Threading.Thread.Sleep(1000);
                                    //Console.WriteLine("队列末运行,执行失败");
                                }
                            }
                            //socket 断开异常
                            catch (System.IO.EndOfStreamException)
                            {
                                if (channel.IsOpen)
                                {
                                    consurment.OnCancel();
                                    consurment = new QueueingBasicConsumer(channel);
                                    channel.BasicConsume(queue, true, consurment);
                                    Console.WriteLine("读取队列自动恢复.......");
                                }
                                else
                                {

                                }

                            }

                        }
                    }
                }
            });
            #endregion

            // RabbitMqClient client0 = new RabbitMqClient("host=192.168.2.110;username=rabbitmq;password=rabbitmq", SerializerType.Json);


        }

        #region Exchange类型处理
        /// <summary> 
        /// Topic 在邦定到队列时必须增加 RowKey属性
        /// 将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。符号“#”匹配一个或多个词，符号“*”匹配不多不少一个词。
        /// </summary>
        public void Topic类型(IModel channel)
        {
            //创建exchange
            channel.ExchangeDeclare("INCOM.DGSExchange", ExchangeType.Topic);

            //创建邦定队列
            var properties = channel.QueueDeclare("INCOM.DGS1", false, false, false, null);
            //将队列邦定到exchange上
            channel.QueueBind("INCOM.DGS1", "INCOM.DGSExchange", "DGS.*");

        }
        /// <summary> 
        /// Direct 处理路由键。需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。这是一个完整的匹配
        /// </summary>
        public void Direct类型(IModel channel)
        {
            //创建exchange
            channel.ExchangeDeclare("INCOM.DGSExchange.Direct", ExchangeType.Direct);

            //创建邦定队列
            var properties = channel.QueueDeclare("INCOM.DGS1", false, false, false, null);
            //将队列邦定到exchange上
            channel.QueueBind("INCOM.DGS1", "INCOM.DGSExchange", "DGS.RealData");

        }

        /// <summary> 
        /// Fanout
        /// 不处理路由键。你只需要简单的将队列绑定到交换机上。一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上
        /// </summary>
        public void Fanout类型(IModel channel)
        {
            //创建exchange
            //INCOM.DGSExchange 类型不能重复创建不然报错
            channel.ExchangeDeclare("INCOM.DGSExchange.Fanout", ExchangeType.Fanout);

            //创建邦定队列
            var properties = channel.QueueDeclare("INCOM.DGS1", false, false, false, null);
            //将队列邦定到exchange上
            channel.QueueBind("INCOM.DGS1", "INCOM.DGSExchange", "DGS.RealData");

        }
        #endregion


        /// <summary>
        /// 联接MQ
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void button_Click(object sender, RoutedEventArgs e)
        {
            示例数据();
        }
        /// <summary>
        /// blockCollectionT处理
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void button1_Click(object sender, RoutedEventArgs e)
        {
            BlockingCollection<int> coll = new BlockingCollection<int>();

            System.Threading.Thread thr = new System.Threading.Thread(() =>
            {
                int s;
                coll.TryTake(out s, 99999);

            });
            thr.Start();

            System.Threading.Thread thr1 = new System.Threading.Thread(() =>
            {
                System.Threading.Thread.Sleep(10000);
                int s;
                coll.TryAdd(1);

            });
            thr1.Start();
        }
    }


    //提供int 与string两种KEY的终端类型支持
    /// <summary>
    /// 终端在线管理
    /// 提供值类型的终端上下线状态管理
    /// 提供String类型的终端上下线状态管理
    /// </summary>
    public class TerminalOnLineManager
    {

        /// <summary>
        /// Key数字类型的终端如808终端,x68的终端
        /// </summary>
        System.Collections.Concurrent.ConcurrentDictionary<long, TerminalEntity> TerDicNum = new System.Collections.Concurrent.ConcurrentDictionary<long, TerminalEntity>();
        /// <summary>
        /// Key字符串类型的终端如OTA终端,国标,地标
        /// </summary>
        System.Collections.Concurrent.ConcurrentDictionary<string, TerminalEntity> TerDicStr = new System.Collections.Concurrent.ConcurrentDictionary<string, TerminalEntity>();

        /// <summary>
        /// Key为终端号,VIN的字符串为Key
        /// </summary>
        System.Collections.Concurrent.ConcurrentDictionary<string, TerminalEntity> TerDic = new System.Collections.Concurrent.ConcurrentDictionary<string, TerminalEntity>();

        /// <summary>
        /// 添加数字类型的终端
        /// </summary>
        /// <param name="ent"></param>
        /// <returns></returns>

        public bool AddNumberTerminal(TerminalEntity ent)
        {

            return true;
        }




    }

    /// <summary>
    /// 终端实体
    /// </summary>
    public class TerminalEntity
    {
        /// <summary>
        /// 终端编号
        /// </summary>
        public long TerminalCode { get; set; }


        /// <summary>
        /// 终端识别码字符串类型
        /// </summary>
        public string Vin { get; set; }

        /// <summary>
        /// 
        /// </summary>
        public bool IsStringFlag { get; set; }
    }
}
